#include "carmela_sdk.h"

struct MQSdkContent
{
	unsigned char		tcp_nodelay ;
	
	char			broker_ip[ 40 + 1 ] ;
	uint16_t		broker_port ;
	
	struct AppendBuffer	*comm_data ;
	struct AppendBuffer	*comm_data_confirm ;
	
	struct AppendBuffer	*tags ;
	struct AppendBuffer	*msg_data ;
	
	int			sock ;
	struct timeval		timeout ;
} ;

void MQSetLogEnv( int log_level , char *format , ... )
{
	va_list		valist ;
	
	va_start( valist , format );
	SetLogFileV( format , valist );
	va_end( valist );
	
	SetLogLevel( log_level );
	
	return;
}

struct MQSdkContent *MQAllocContent( char *broker_ip , uint16_t broker_port )
{
	struct MQSdkContent	*sdk = NULL ;
	char			*env_MQ_TCP_NODEALY = NULL ;
	char			*env_MQ_SDK_TCP_NODEALY = NULL ;
	
	sdk = (struct MQSdkContent *)malloc( sizeof(struct MQSdkContent) ) ;
	if( sdk == NULL )
		return NULL;
	memset( sdk , 0x00 , sizeof(struct MQSdkContent) );
	
	env_MQ_TCP_NODEALY = getenv( "MQ_TCP_NODELAY" ) ;
	env_MQ_SDK_TCP_NODEALY = getenv( "MQ_SDK_TCP_NODELAY" ) ;
	if( ( env_MQ_TCP_NODEALY && atoi(env_MQ_TCP_NODEALY) ) || ( env_MQ_SDK_TCP_NODEALY && atoi(env_MQ_SDK_TCP_NODEALY) ) )
	{
		sdk->tcp_nodelay = 1 ;
	}
	
	strncpy( sdk->broker_ip , broker_ip , sizeof(sdk->broker_ip)-1 );
	sdk->broker_port = broker_port ;
	
	sdk->comm_data = AllocBufferInitSize( 1024 ) ;
	if( sdk->comm_data == NULL )
	{
		MQFreeContent( sdk );
		return NULL;
	}
	
	sdk->comm_data_confirm = AllocBufferInitSize( 1024 ) ;
	if( sdk->comm_data_confirm == NULL )
	{
		MQFreeContent( sdk );
		return NULL;
	}
	
	sdk->tags = AllocBufferInitSize( 1024 ) ;
	if( sdk->tags == NULL )
	{
		MQFreeContent( sdk );
		return NULL;
	}
	
	sdk->msg_data = AllocBufferInitSize( 1024 ) ;
	if( sdk->msg_data == NULL )
	{
		MQFreeContent( sdk );
		return NULL;
	}
	
	sdk->sock = -1 ;
	
	return sdk;
}

void MQFreeContent( struct MQSdkContent *sdk )
{
	if( sdk )
	{
		if( sdk->comm_data )
			FreeBuffer( sdk->comm_data );
		
		if( sdk->comm_data_confirm )
			FreeBuffer( sdk->comm_data_confirm );
		
		if( sdk->tags )
			FreeBuffer( sdk->tags );
		
		if( sdk->msg_data )
			FreeBuffer( sdk->msg_data );
		
		free( sdk );
	}
	
	return;
}

static int SendRequestToBrokerAndReceiveResponse( struct MQSdkContent *sdk )
{
	struct TDHBContext	tdhb ;
	
	int			nret = 0 ;
	
	memset( & tdhb , 0x00 , sizeof(struct TDHBContext) );
	nret = TDHBSendData( sdk->sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , & tdhb , & (sdk->timeout) ) ;
	if( nret )
	{
		ERRORLOG( "TDHBSendData failed[%d] , sent_len[%zu]" , nret , tdhb.body_len-tdhb.body_remain_len )
		return nret;
	}
	else
	{
		INFOLOG( "TDHBSendData ok , sent_len[%zu]" , tdhb.body_len-tdhb.body_remain_len )
	}
	
	CleanAppendBuffer( sdk->comm_data );
	memset( & tdhb , 0x00 , sizeof(struct TDHBContext) );
	
_GOTO_RECEIVE_AGAIN :
	nret = TDHBReceiveData( sdk->sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , & tdhb , & (sdk->timeout) ) ;
	if( nret == TDHB_ERROR_RECV_BUFFER_OVERFLOW )
	{
		INFOLOG( "TDHBReceiveData failed[TDHB_ERROR_RECV_BUFFER_OVERFLOW] , comm_data buf[%p] bufs_size[%zu] , errno[%d]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , errno )
		
		nret = ResizeAppendBuffer( sdk->comm_data , tdhb.body_len+1 ) ;
		if( nret )
		{
			ERRORLOG( "ResizeAppendBuffer failed , comm_data buf[%p] buf_size[%zu] , errno[%d]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , errno )
			return MQSDK_ERROR_RESIZE_BUFFER;
		}
		else
		{
			DEBUGLOG( "ResizeAppendBuffer ok , new comm_data buf[%p] buf_size[%d]bytes" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) )
			goto _GOTO_RECEIVE_AGAIN;
		}
	}
	else if( nret )
	{
		OffsetAppendBufferFillPtr( sdk->comm_data , tdhb.body_len );
		ERRORLOG( "TDHBReceiveData failed[%d] , comm_data buf[%p] buf_len[%zu/%zu]" , nret , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) )
		return MQSDK_ERROR_RECEIVE;
	}
	else
	{
		OffsetAppendBufferFillPtr( sdk->comm_data , tdhb.body_len );
		INFOLOG( "TDHBReceiveData ok , comm_data buf[%p] buf_len[%zu]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) )
	}
	
	return 0;
}

static int RequestBroker( struct MQSdkContent *sdk )
{
	int		old_sock ;
	
	int		nret = 0 ;
	
	old_sock = sdk->sock ;
	
	if( sdk->sock < 0 )
	{
		nret = MQConnect( sdk ) ;
		if( nret )
		{
			ERRORLOG( "MQConnect failed[%d]" , nret )
			return nret;
		}
		else
		{
			DEBUGLOG( "MQConnect ok" )
		}
	}
	
	sdk->timeout.tv_sec = 30 ;
	nret = SendRequestToBrokerAndReceiveResponse( sdk ) ;
	if( nret )
	{
		ERRORLOG( "SendRequestToBrokerAndReceiveResponse failed[%d]" , nret )
		if( old_sock < 0 )
		{
			DEBUGLOG( "MQDisconnect" )
			MQDisconnect( sdk );
		}
		
		return MQSDK_ERROR_SEND;
	}
	else
	{
		INFOLOG( "SendRequestToBrokerAndReceiveResponse ok" )
	}
	
	if( old_sock < 0 )
	{
		DEBUGLOG( "MQDisconnect" )
		MQDisconnect( sdk );
	}
	
	return 0;
}

static int ReceiveRequestFromBrokerAndSendConsumeMessageResponse( struct MQSdkContent *sdk )
{
	struct TDHBContext	tdhb ;
	
	int			nret = 0 ;
	
	CleanAppendBuffer( sdk->comm_data );
	memset( & tdhb , 0x00 , sizeof(struct TDHBContext) );
	
_GOTO_RECEIVE_AGAIN :
	nret = TDHBReceiveData( sdk->sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , & tdhb , & (sdk->timeout) ) ;
	if( nret == TDHB_ERROR_RECV_BUFFER_OVERFLOW )
	{
		INFOLOG( "TDHBReceiveData failed[TDHB_ERROR_RECV_BUFFER_OVERFLOW] , comm_data buf[%p] bufs_size[%zu] , errno[%d]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , errno )
		
		nret = ResizeAppendBuffer( sdk->comm_data , tdhb.body_len+1 ) ;
		if( nret )
		{
			ERRORLOG( "ResizeAppendBuffer failed , comm_data buf[%p] buf_size[%zu] , errno[%d]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) , errno )
			return MQSDK_ERROR_RESIZE_BUFFER;
		}
		else
		{
			DEBUGLOG( "ResizeAppendBuffer ok , new comm_data buf[%p] buf_size[%d]bytes" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) )
			goto _GOTO_RECEIVE_AGAIN;
		}
	}
	else if( nret )
	{
		OffsetAppendBufferFillPtr( sdk->comm_data , tdhb.body_len );
		ERRORLOG( "TDHBReceiveData failed[%d] , comm_data buf[%p] buf_len[%zu/%zu]" , nret , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , GetAppendBufferSize(sdk->comm_data) )
		return MQSDK_ERROR_RECEIVE;
	}
	else
	{
		OffsetAppendBufferFillPtr( sdk->comm_data , tdhb.body_len );
		INFOLOG( "TDHBReceiveData ok , comm_data buf[%p] buf_len[%zu]" , GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) )
	}
	
	return 0;
}

int MQCreateStore( struct MQSdkContent *sdk , char *store_name )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CREATESTORE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND CREATE STORE REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_CREATESTORE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	INFOLOG( "RECV CREATE STORE RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQDestroyStore( struct MQSdkContent *sdk , char *store_name )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_DESTROYSTORE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND DESTROY STORE REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_DESTROYSTORE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	INFOLOG( "RECV DESTROY STORE RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQCreateTopic( struct MQSdkContent *sdk , char *store_name , char *topic_name )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint8_t			topic_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CREATETOPIC_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	len = strlen(topic_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "topic_name[%s] too long" , topic_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	topic_name_len = (uint8_t)len ;
	MEMPACK_TOPIC_NAME_LEN( return -1 , topic_name_len , sdk->comm_data )
	MEMPACK_TOPIC_NAME( return -1 , topic_name , topic_name_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND CREATE TOPIC REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_CREATETOPIC_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	INFOLOG( "RECV CREATE TOPIC RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQDestroyTopic( struct MQSdkContent *sdk , char *store_name , char *topic_name )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint8_t			topic_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_DESTROYTOPIC_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	len = strlen(topic_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "topic_name[%s] too long" , topic_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	topic_name_len = (uint8_t)len ;
	MEMPACK_TOPIC_NAME_LEN( return -1 , topic_name_len , sdk->comm_data )
	MEMPACK_TOPIC_NAME( return -1 , topic_name , topic_name_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND DESTROY TOPIC REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_DESTROYTOPIC_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	INFOLOG( "RECV SUBSCRIBE MESSAGE RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQProduceMessage( struct MQSdkContent *sdk , char *store_name , char *topic_name , char *tags , char *msg_data , uint64_t *p_msg_offset )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint8_t			topic_name_len ;
	uint16_t		tags_len ;
	uint32_t		msg_data_len ;
	uint16_t		result ;
	uint64_t		msg_offset ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_PRODUCEMESSAGE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	len = strlen(topic_name) ;
	if( len > 0xFFFF )
	{
		ERRORLOG( "topic_name[%s] too long" , topic_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	topic_name_len = (uint8_t)len ;
	MEMPACK_TOPIC_NAME_LEN( return -1 , topic_name_len , sdk->comm_data )
	MEMPACK_TOPIC_NAME( return -1 , topic_name , topic_name_len , sdk->comm_data )
	
	len = strlen(tags) ;
	if( len > 0xFFFF )
	{
		ERRORLOG( "tags[%s] too long" , tags )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	tags_len = (uint16_t)len ;
	MEMPACK_TAGS_LEN( return -1 , tags_len , sdk->comm_data )
	MEMPACK_TAGS( return -1 , tags , tags_len , sdk->comm_data )
	
	len = strlen(msg_data) ;
	if( len > 0xFFFFFFFF )
	{
		ERRORLOG( "msg_data[%.20s...] too long" , msg_data )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	msg_data_len = (uint32_t)len ;
	MEMPACK_MSG_DATA_LEN( return -1 , msg_data_len , sdk->comm_data )
	MEMPACK_MSG_DATA( return -1 , msg_data , msg_data_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND PRODUCE MESSAGE REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_PRODUCEMESSAGE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	MEMUNPACK_MSG_OFFSET( return nret , sdk->comm_data , msg_offset )
	if( p_msg_offset )
		(*p_msg_offset) = msg_offset ;
	
	INFOLOG( "RECV PRODUCE MESSAGE RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQBeginSubscribeMessage( struct MQSdkContent *sdk , char *store_name , char *topic_name , char *subscribe_group_name )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint8_t			topic_name_len ;
	char			unique_subscribe_group_name[ MQ_SUBSCRIBE_GROUP_NAME_MAX_LEN+1 ] ;
	uint8_t			subscribe_group_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_SUBSCRIBEMESSAGE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	len = strlen(topic_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "topic_name[%s] too long" , topic_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	topic_name_len = (uint8_t)len ;
	MEMPACK_TOPIC_NAME_LEN( return -1 , topic_name_len , sdk->comm_data )
	MEMPACK_TOPIC_NAME( return -1 , topic_name , topic_name_len , sdk->comm_data )
	
	if( subscribe_group_name == NULL )
	{
		int		tid ;
		uint64_t	u64 ;
		
		memset( unique_subscribe_group_name , 0x00 , sizeof(unique_subscribe_group_name) );
		
		nret = GetNetInterfaceIp( 1 , NULL , unique_subscribe_group_name , NULL ) ;
		if( nret )
		{
			ERRORLOG( "GetNetInterfaceIp failed[%d]" , nret )
			return MQSDK_ERROR_GET_IP;
		}
		
		unique_subscribe_group_name[8] = '-' ;
		
		tid = syscall(SYS_gettid) ;
		if( tid == -1 )
		{
			ERRORLOG( "syscall(SYS_gettid) failed[%d]" , nret )
			return MQSDK_ERROR_SYSCALL;
		}
		
		u64 = tid ;
		u64 = hton64( u64 ) ;
		bytes2hexstr( (unsigned char*) & u64 , sizeof(u64) , (unsigned char *)(unique_subscribe_group_name+9) );
		
		subscribe_group_name = unique_subscribe_group_name ;
		DEBUGLOG( "unique_subscribe_group_name[%s]" , unique_subscribe_group_name )
	}
	len = strlen(subscribe_group_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "subscribe_group_name[%s] too long" , subscribe_group_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	subscribe_group_name_len = (uint8_t)len ;
	MEMPACK_SUBSCRIBE_GROUP_NAME_LEN( return -1 , subscribe_group_name_len , sdk->comm_data )
	MEMPACK_SUBSCRIBE_GROUP_NAME( return -1 , subscribe_group_name , subscribe_group_name_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND SUbSCRIBE MESSAGE REQUEST" )
	
	DEBUGLOG( "MQConnect ..." )
	nret = MQConnect( sdk ) ;
	if( nret )
	{
		ERRORLOG( "MQConnect failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "MQConnect ok" )
	}
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		MQDisconnect( sdk );
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		MQDisconnect( sdk );
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_SUBSCRIBEMESSAGE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		MQDisconnect( sdk );
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	INFOLOG( "RECV SUBSCRIBE MESSAGE RESPONSE - result[%"PRIu16"]" , result )
	if( result )
	{
		ERRORLOG( "result[%"PRIu16"]" , result )
		MQDisconnect( sdk );
	}
	else
	{
		INFOLOG( "result[%"PRIu16"]" , result )
	}
	
	return result;
}

int MQConsumeMessage( struct MQSdkContent *sdk , char **pp_topic , char **pp_tags , uint32_t *p_msg_data_len , char **pp_msg_data , uint64_t *p_topic_offset )
{
	union MQProcessCode	process_code ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint16_t		tags_len ;
	char			*tags = NULL ;
	uint32_t		msg_data_len ;
	char			*msg_data = NULL ;
	uint8_t			comm_version ;
	uint64_t		topic_offset ;
	
	struct TDHBContext	tdhb ;
	
	int			nret = 0 ;
	
_GOTO_RECEIVE_AGAIN :
	nret = ReceiveRequestFromBrokerAndSendConsumeMessageResponse( sdk ) ;
	if( nret )
	{
		ERRORLOG( "ReceiveRequestFromBrokerAndSendConsumeMessageResponse failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "ReceiveRequestFromBrokerAndSendConsumeMessageResponse ok" )
	}
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_CONSUMEMESSAGE_REQUEST) )
	{
		INFOLOG( "RECV CONSUME MESSAGE REQUEST" )
		
		MEMUNPACK_TOPIC_NAME_LEN( return nret , sdk->comm_data , topic_name_len )
		MEMUNPACK_TOPIC_NAME( return nret , sdk->comm_data , topic_name_len , topic_name )
		
		MEMUNPACK_TAGS_LEN( return nret , sdk->comm_data , tags_len )
		MEMUNPACK_TAGS( return nret , sdk->comm_data , tags_len , tags )
		
		MEMUNPACK_MSG_DATA_LEN( return nret , sdk->comm_data , msg_data_len )
		MEMUNPACK_MSG_DATA( return nret , sdk->comm_data , msg_data_len , msg_data )
		
		MEMUNPACK_TOPIC_OFFSET( return nret , sdk->comm_data , topic_offset )
		
		CleanAppendBuffer( sdk->comm_data_confirm );
		
		comm_version = MQ_COMM_PROTO_VERSION_1 ;
		MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data_confirm )
		
		process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CONSUMEMESSAGE_RESPONSE ;
		MEMPACK_PROCESS_CODE( return nret , process_code , sdk->comm_data_confirm )
		
		if( nret )
		{
			MEMPACK_RESULT( return nret , MQSDK_ERROR_MESSAGE_TOO_SHORT , sdk->comm_data_confirm )
		}
		else
		{
			MEMPACK_RESULT( return nret , MQSDK_SUCCESS , sdk->comm_data_confirm )
		}
		
		INFOLOG( "SEND CONSUME MESSAGE RESPONSE" )
		
		memset( & tdhb , 0x00 , sizeof(struct TDHBContext) );
		nret = TDHBSendData( sdk->sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(sdk->comm_data_confirm) , GetAppendBufferLength(sdk->comm_data_confirm) , & tdhb , & (sdk->timeout) ) ;
		if( nret )
		{
			ERRORLOG( "TDHBSendData failed[%d] , sent_len[%zu]" , nret , tdhb.body_len-tdhb.body_remain_len )
			return nret;
		}
		else
		{
			INFOLOG( "TDHBSendData ok , sent_len[%zu]" , tdhb.body_len-tdhb.body_remain_len )
		}
		
		if( pp_topic )
			(*pp_topic) = topic_name ;
		if( pp_tags )
			(*pp_tags) = tags ;
		if( p_msg_data_len )
			(*p_msg_data_len) = msg_data_len ;
		if( pp_msg_data )
			(*pp_msg_data) = msg_data ;
		if( p_topic_offset )
			(*p_topic_offset) = topic_offset ;
	}
	else if( process_code.u32 == *(uint32_t*)(MQ_PROCESSCODE_HEARTBEAT_REQUEST) )
	{
		INFOLOG( "RECV HEARTBEAT MESSAGE REQUEST" )
		
		CleanAppendBuffer( sdk->comm_data_confirm );
		memset( & tdhb , 0x00 , sizeof(struct TDHBContext) );
		
		comm_version = MQ_COMM_PROTO_VERSION_1 ;
		MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data_confirm )
		
		process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_HEARTBEAT_RESPONSE ;
		MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data_confirm )
		
		MEMPACK_RESULT( return nret , MQSDK_SUCCESS , sdk->comm_data_confirm )
		
		INFOLOG( "SEND HEARTBEAT RESPONSE" )
		
		nret = TDHBSendData( sdk->sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(sdk->comm_data_confirm) , GetAppendBufferLength(sdk->comm_data_confirm) , & tdhb , & (sdk->timeout) ) ;
		if( nret )
		{
			ERRORLOG( "TDHBSendData failed[%d] , sent_len[%zu]" , nret , tdhb.body_len-tdhb.body_remain_len )
			return nret;
		}
		else
		{
			INFOLOG( "TDHBSendData ok , sent_len[%zu]" , tdhb.body_len-tdhb.body_remain_len )
		}
		
		goto _GOTO_RECEIVE_AGAIN;
	}
	else
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	return 0;
}

int MQEndSubscribeMessage( struct MQSdkContent *sdk )
{
	INFOLOG( "MQDisconnect" )
	MQDisconnect( sdk );
	
	return 0;
}

int MQProduceDelayMessage( struct MQSdkContent *sdk , uint32_t delay_seconds , char *store_name , char *topic_name , char *tags , char *msg_data , uint64_t *p_delay_msg_offset )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint8_t			topic_name_len ;
	uint16_t		tags_len ;
	uint32_t		msg_data_len ;
	uint16_t		result ;
	uint64_t		delay_msg_offset ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_PRODUCEDELAYMESSAGE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	MEMPACK_DELAY_SECONDS( return -1 , delay_seconds , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	len = strlen(topic_name) ;
	if( len > 0xFFFF )
	{
		ERRORLOG( "topic_name[%s] too long" , topic_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	topic_name_len = (uint8_t)len ;
	MEMPACK_TOPIC_NAME_LEN( return -1 , topic_name_len , sdk->comm_data )
	MEMPACK_TOPIC_NAME( return -1 , topic_name , topic_name_len , sdk->comm_data )
	
	len = strlen(tags) ;
	if( len > 0xFFFF )
	{
		ERRORLOG( "tags[%s] too long" , tags )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	tags_len = (uint16_t)len ;
	MEMPACK_TAGS_LEN( return -1 , tags_len , sdk->comm_data )
	MEMPACK_TAGS( return -1 , tags , tags_len , sdk->comm_data )
	
	len = strlen(msg_data) ;
	if( len > 0xFFFFFFFF )
	{
		ERRORLOG( "msg_data[%.20s...] too long" , msg_data )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	msg_data_len = (uint32_t)len ;
	MEMPACK_MSG_DATA_LEN( return -1 , msg_data_len , sdk->comm_data )
	MEMPACK_MSG_DATA( return -1 , msg_data , msg_data_len , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND PRODUCE DELAY MESSAGE REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_PRODUCEDELAYMESSAGE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	MEMUNPACK_DELAY_MSG_OFFSET( return nret , sdk->comm_data , delay_msg_offset )
	if( p_delay_msg_offset )
		(*p_delay_msg_offset) = delay_msg_offset ;
	
	INFOLOG( "RECV PRODUCE DELAY MESSAGE RESPONSE - result[%"PRIu16"] delay_msg_offset[%"PRIu64"]" , result , (*p_delay_msg_offset) )
	
	return result;
}

int MQCancelDelayMessage( struct MQSdkContent *sdk , char *store_name , uint64_t delay_msg_offset )
{
	ssize_t			len ;
	uint8_t			comm_version ;
	union MQProcessCode	process_code ;
	uint8_t			store_name_len ;
	uint16_t		result ;
	
	int			nret = 0 ;
	
	ResetAppendBufferFillPtr( sdk->comm_data );
	
	comm_version = MQ_COMM_PROTO_VERSION_1 ;
	MEMPACK_COMM_VERSION( return nret , comm_version , sdk->comm_data )
	
	process_code.u32 = *(uint32_t*)MQ_PROCESSCODE_CANCELDELAYMESSAGE_REQUEST ;
	MEMPACK_PROCESS_CODE( return -1 , process_code , sdk->comm_data )
	
	len = strlen(store_name) ;
	if( len > 0xFF )
	{
		ERRORLOG( "store_name[%s] too long" , store_name )
		return MQSDK_ERROR_LENGTH_FIELD_TOO_BIG;
	}
	store_name_len = (uint8_t)len ;
	MEMPACK_STORE_NAME_LEN( return -1 , store_name_len , sdk->comm_data )
	MEMPACK_STORE_NAME( return -1 , store_name , store_name_len , sdk->comm_data )
	
	MEMPACK_DELAY_MSG_OFFSET( return -1 , delay_msg_offset , sdk->comm_data )
	
	DEBUGHEXLOG( GetAppendBufferBase(sdk->comm_data) , GetAppendBufferLength(sdk->comm_data) , "pack ok , [%"PRIu32"]bytes" , GetAppendBufferLength(sdk->comm_data) )
	INFOLOG( "SEND CANCEL DELAY MESSAGE REQUEST" )
	
	DEBUGLOG( "RequestBroker ..." )
	nret = RequestBroker( sdk ) ;
	if( nret )
	{
		ERRORLOG( "RequestBroker failed[%d]" , nret )
		return nret;
	}
	else
	{
		INFOLOG( "RequestBroker ok" )
	}
	
	ResetAppendBufferParsePtr( sdk->comm_data );
	
	MEMUNPACK_COMM_VERSION( return -1 , sdk->comm_data , comm_version )
	if( comm_version != MQ_COMM_PROTO_VERSION_1 )
	{
		ERRORLOG( "invalid comm_version[%c][0x%02X]" , comm_version , comm_version )
		return MQSDK_ERROR_VERSION_INVALID;
	}
	
	memset( & process_code , 0x00 , sizeof(union MQProcessCode) );
	MEMUNPACK_PROCESS_CODE( return -1 , sdk->comm_data , process_code )
	if( process_code.u32 != *(uint32_t*)(MQ_PROCESSCODE_CANCELDELAYMESSAGE_RESPONSE) )
	{
		ERRORLOG( "unmatch process_code[%.4s][0x%02X 0x%02X 0x%02X 0x%02X]" , process_code.str , process_code.str[0] , process_code.str[1] , process_code.str[2] , process_code.str[3] )
		return MQSDK_ERROR_PROCESS_CODE_INVALID;
	}
	
	MEMUNPACK_RESULT( return nret , sdk->comm_data , result )
	
	INFOLOG( "RECV CANCEL DELAY MESSAGE RESPONSE - result[%"PRIu16"]" , result )
	
	return result;
}

int MQConnect( struct MQSdkContent *sdk )
{
	if( sdk->sock >= 0 )
	{
		ERRORLOG( "sock[%d] connected already" , sdk->sock )
		return MQSDK_ERROR_INTERNAL;
	}
	
	sdk->sock = TDHBConnect( sdk->broker_ip , sdk->broker_port ) ;
	if( sdk->sock < 0 )
	{
		ERRORLOG( "TDHBConnect failed[%d] , broker[%s:%d] , errno[%d]" , sdk->sock , sdk->broker_ip , sdk->broker_port , errno )
		return MQSDK_ERROR_CONNECT;
	}
	else
	{
		INFOLOG( "TDHBConnect ok , broker[%s:%d] , sock[%d]" , sdk->broker_ip , sdk->broker_port , sdk->sock )
	}
	
	if( sdk->tcp_nodelay )
	{
		int	onoff = 1 ;
		setsockopt( sdk->sock , IPPROTO_TCP , TCP_NODELAY , (void*) & onoff , sizeof(int) );
		INFOLOG( "enable socket TCP_NODELAY" )
	}
	
	return 0;
}

int MQDisconnect( struct MQSdkContent *sdk )
{
	if( sdk->sock < 0 )
	{
		ERRORLOG( "sock[%d] not connected" , sdk->sock )
		return MQSDK_ERROR_INTERNAL;
	}
	
	DEBUGLOG( "TDHBDisconnect" )
	TDHBDisconnect( sdk->sock );
	sdk->sock = -1 ;
	
	return 0;
}

